package org.hbhk.hms.mq;

import java.util.Map;

import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import com.google.common.collect.Maps;

@Service
public class MqConsumer implements InitializingBean {

	protected static final Logger logger = LoggerFactory.getLogger(MqConsumer.class);

	public static Map<String, DefaultMQPushConsumer> mqProducerCache = Maps.newConcurrentMap();

	@Value("${mq.namesrvAddr}")
	private  String namesrvAddr;

	@Value("${mq.consumeThread:10}")
	public  Integer consumeThread = 10;

	public  DefaultMQPushConsumer get(String mqName) {

		if (mqProducerCache.containsKey(mqName)) {
			return mqProducerCache.get(mqName);
		}

		synchronized (logger) {
			if (!mqProducerCache.containsKey(mqName)) {
				if (StringUtils.isNotEmpty(mqName)) {

					DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(mqName);

					consumer.setNamesrvAddr(namesrvAddr);

					consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
					if (consumeThread == null) {
						consumeThread = 10;
					}
					consumer.setConsumeThreadMin(consumeThread / 2);
					consumer.setConsumeThreadMax(consumeThread);

					mqProducerCache.put(mqName, consumer);

				}

			}
		}
		return mqProducerCache.get(mqName);
	}

	public void afterPropertiesSet() throws Exception {

	}

}
